#include <stdio.h>
#include <pthread.h>
#include <unistd.h>
#include <stdlib.h>
#include <string.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/socket.h>

#include "socket.h"
#include "pthreadpool.h"
#include "epoll.h"
#include "data_packet.h"
#include "filedispat.h"
#include "like.h"
#include "fidata_list.h"

extern void Judge_fileinfo_or_chatinfo(Conndata *data);

void *start_routine(void *arg)
{
	threadpool *pool = (threadpool*)arg;
	Task process_task;
	while(1){
		pthread_mutex_lock(&pool->mutex);
		while(pool->queue_size == 0 && pool->shutdown == THREADPOOL_ON){
			pthread_cond_wait(&pool->cond,&pool->mutex);
		}
		if(pool->shutdown ==THREADPOOL_OFF){
			pthread_mutex_unlock(&pool->mutex);
			printf("threadpool 1 close ID: 0x%lx\n",pthread_self());
			break;
		}
		process_task = (pool->Task_queue[pool->Frount]);
		pool->Task_queue[pool->Frount].func = NULL;
		pool->Task_queue[pool->Frount].arg = NULL;
		pool->Frount = ((pool->Frount+1)%(pool->queue_max));
		pool->queue_size--;
		pthread_cond_broadcast(&pool->Task_add_cond);
		printf("thread: 0x%lxdispose work!\n",pthread_self());
		pthread_mutex_unlock(&pool->mutex);
		process_task.func(process_task.arg);
	}
	pthread_exit(NULL);
}

threadpool *threadpool_create(int pthread_max,int queue_max)
{
	int ret;
	threadpool *pool;
	pool = (threadpool*)malloc(sizeof(struct threadpool));
	if(pool == NULL){
		fprintf(stderr,"pool create error!");
		return NULL;
	}
	do{
		pool->queue_size = 0;
		pool->Frount = 0;
		pool->Rear = -1;
		pool->queue_max = queue_max;
		pool->shutdown = THREADPOOL_ON;
		pool->pthread_max = pthread_max;
		pool->Frount = pool->Rear = 0;
		pool->pidID = (pthread_t *)malloc(sizeof(threadpool)*pthread_max);
		if(pool->pidID == NULL){
			fprintf(stderr,"pidID create error!\n");
			free(pool->pidID);
			break;
		}
		if( pthread_mutex_init(&pool->mutex,NULL)!=0||
			pthread_mutex_init(&pool->link_lock,NULL)!=0||
			pthread_cond_init(&pool->cond,NULL)!=0 || 
			pthread_cond_init(&pool->Task_add_cond,NULL)!=0 )
		{
			fprintf(stderr,"init the lock or cond fail\n");
			break;
		}
		pool->Task_queue = (Task *)malloc(sizeof(Task)*queue_max);
		if(pool->Task_queue == NULL){
			perror("Task_queue\n");
			break;
		}

		for(int i=0;i<pthread_max;i++){
			//创建任务线程！
			ret = pthread_create(&pool->pidID[i], NULL,start_routine, (void *)pool);
			if(ret){
				perror("pthread_create!\n");
				break;
			}		
		}
		if(ret)
			break;
		return pool;		
	}while(0);

	fprintf(stderr,"threadpool create error!\n");
	free(pool);
	return NULL;
}

void Connect_procfunc(void *arg)
{
	char str[64];
	int ret,clifd,flag;
	struct epoll_event epoll;
	Conndata *data = (Conndata *)arg;
	struct sockaddr_in cli_addr;
	socklen_t client = sizeof(socklen_t);
	clifd = accept(data->event_fd,(struct sockaddr*)&cli_addr,&client);
	flag = fcntl(clifd, F_GETFL);       /* 修改connfd为非阻塞读 */    
    flag |= O_NONBLOCK;
    fcntl(clifd, F_SETFL, flag);
	
	printf("received from %s at PORT %d\n",
		inet_ntop(AF_INET, &cli_addr.sin_addr, str, sizeof(str)),
		ntohs(cli_addr.sin_port));
	printf("cfd %d---client\n", clifd);

	ret = eproot_add_fd(clifd,data->efd_size,epoll);
	if(ret ==-1)
		perr_exit("epoll_ctl error");
	free(data);
}

void Message_procfunc(void *arg)
{
	Conndata *data = (Conndata *)arg; 
	int n,mark,ret;
	char buf[SOCK_BUF_SIZE];
	if((n = readn(data->event_fd,buf,sizeof(buf)))==-1)
			perror("recv=0");
			
	sscanf(buf,"%d ",&mark);
	if(mark == 0){
		pthread_mutex_lock(&data->pool->link_lock);
		ret = register_operation(data->event_fd,buf,user,head);
		if(ret==-1){
			send_register_erron(data->event_fd);	
		}else{
			send_socket(data->event_fd,head);
			}
		pthread_mutex_unlock(&data->pool->link_lock);
	}else if(mark == 1){
		//printf("----------------------------\n");
		public_catinfo_post(data->event_fd,public_cat,buf);
	}else if(mark == 2){
		private_catinfo_post(data->event_fd,packet,head,buf);
	}
}
	

void File_Message_procfunc(void *arg)
{
	int target_fd,mark,ret;
	char buf[SOCK_BUF_SIZE];
	char user[64];
	filestruct file_data;
	Conndata *data = (Conndata *)arg;
	mark = read_fifo_mark(data->event_fd);
	if(mark == 5){
		memset(buf,0,SOCK_BUF_SIZE);
		readn(data->event_fd,buf,SOCK_BUF_SIZE);
		printf("%s\n\n",buf);
		recv_file_data(data->event_fd, buf,&file_data,user);
		target_fd = fd_ergodic_list(file_data.target_user,head);
		if(target_fd == -1){
			send_file_error(data->event_fd);
			return ;
		}
		memset(buf,0,sizeof(SOCK_BUF_SIZE));
		fidata_node senddata;
		strcpy(senddata.name,fd_find_name(target_fd,head));
		printf("%s\n",user);
		sprintf(buf,"%d%s %s %d",mark,user,file_data.filename,file_data.fd_size);
		printf("$$$$$$$$$$%s￥￥￥￥￥￥￥￥￥\n",buf);
		writen(target_fd, buf, SOCK_BUF_SIZE);
	}else if(mark == 6){
		memset(buf,0,SOCK_BUF_SIZE);
		ret = server_transfer_data(data,buf);
		if(ret == -1)
			return ;
	}
	return ;	
}

void Epoll_Detection_profunc(void *arg)
{
	char buf[SOCK_BUF_SIZE];
	int iread;
	Conndata *data = (Conndata *)arg;
	iread = recvn_read_only(data->event_fd, buf, SOCK_BUF_SIZE);
	if(iread == 0){
		pthread_mutex_lock(&data->pool->link_lock);
		del_like_user(data->event_fd,head);
		if(on_line_updata(head))
			perr_exit("send error\n");
		epoll_ctl_del(data->event_fd,data->efd_size,head); 
		printf("client[%d] closed connection\n", data->event_fd);
		pthread_mutex_unlock(&data->pool->link_lock);
	}else if(iread == SOCK_BUF_SIZE){
		//printf("#######\n");
		Judge_fileinfo_or_chatinfo(data);
	}else{
		printf("******iread = %d******\n",iread);
		return ;
	}
}

void pthreadpool_Taks_add(threadpool *pool,void (*task_func)(void *),void *arg)
{
	pthread_mutex_lock(&pool->mutex);
	if(pool->queue_size == pool->queue_max)
		pthread_cond_wait(&pool->Task_add_cond,&pool->mutex);
	if(pool->shutdown == THREADPOOL_OFF){
		pthread_mutex_unlock(&pool->mutex);
		printf("threadpool close ID: 0x%lx\n",pthread_self());
		return ;
	}
	pool->Task_queue[pool->Rear].func = task_func;
	pool->Task_queue[pool->Rear].arg = arg;
	pool->Rear = (pool->Rear+1)%(pool->queue_max);
	printf("The %d task is added to the queue\n",pool->Rear);
	//printf("add task %d\n",*(int*)arg);
	pool->queue_size++;
	pthread_cond_signal(&pool->cond);
	pthread_mutex_unlock(&pool->mutex);
}

void pthreadpool_destroy(threadpool *pool)
{
	pool->shutdown = THREADPOOL_OFF;
	pthread_cond_signal(&pool->Task_add_cond);
	for(int i=0;i<=pool->pthread_max;i++)
		pthread_cond_signal(&pool->cond);
	free(pool->pidID);
	free(pool->Task_queue);
	pthread_mutex_destroy(&pool->mutex);
	pthread_cond_destroy(&pool->cond);
	pthread_cond_destroy(&pool->Task_add_cond);
	free(pool);
}


